use crate::component::request::Request;
use crate::component::response::Response;
use crate::component::{ParseResult, Task};
use crate::downloader::Downloader;
use crate::middleware::{DownloaderMiddleware, DownloaderMwErrResult, DownloaderMwResult};
use crate::pipeline::{ItemProduct, Pipeline};
use crate::scheduler::Scheduler;
use crate::spider::Spider;
use log::debug;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

// 一个请求在发送给 downloader 后, 及给 pipeline 前的处理结果
// `download_one()` 的返回值
enum IntermediateProcessResult {
    Request(Request),
    Response(Response),
    Ignore,
}

pub struct EngineBuilder<M, ItemType>
where
    M: Send + Sync + 'static,
    ItemType: Send + Sync + 'static,
{
    downloader: Downloader,
    scheduler: Arc<Mutex<Scheduler<M, ItemType>>>,
    max: usize,
    spider: Option<Arc<dyn Spider<M = M, ItemType = ItemType>>>,
    pipelines: Arc<Vec<tokio::sync::Mutex<Box<dyn Pipeline<ItemType>>>>>,
    dmws: Arc<Vec<tokio::sync::Mutex<Box<dyn DownloaderMiddleware>>>>,
}

impl<M, ItemType> EngineBuilder<M, ItemType>
where
    M: Send + Sync + 'static,
    ItemType: Send + Sync + 'static,
{
    /// spider 必须要提供的, new 出来的 EngineBuilder 中 spider 是 None.
    /// 默认最大并发数 4.
    pub fn new() -> Self {
        Self {
            downloader: Downloader::new(None),
            scheduler: Arc::new(Mutex::new(Scheduler::new())),
            max: 4,
            spider: None,
            pipelines: Arc::new(Vec::new()),
            dmws: Arc::new(Vec::new()),
        }
    }

    pub fn max(&mut self, max: usize) -> &mut Self {
        self.max = max;
        self
    }

    // TODO: 改为 HashMap, 然后内部自己组装为 HeaderMap
    pub fn default_headers(&mut self, headers: HashMap<&'static str, String>) -> &mut Self {
        self.downloader = Downloader::new(Some(headers));
        self
    }

    pub fn spider(&mut self, spider: Arc<dyn Spider<M = M, ItemType = ItemType>>) -> &mut Self {
        self.spider = Some(spider);
        self
    }

    pub fn add_pipeline(&mut self, pipeline: Box<dyn Pipeline<ItemType>>) -> &mut Self {
        Arc::get_mut(&mut self.pipelines)
            .unwrap()
            .push(tokio::sync::Mutex::new(pipeline));
        self
    }

    pub fn add_dmw(&mut self, dmw: Box<dyn DownloaderMiddleware>) -> &mut Self {
        Arc::get_mut(&mut self.dmws)
            .unwrap()
            .push(tokio::sync::Mutex::new(dmw));
        self
    }

    pub fn build(&self) -> Engine<M, ItemType> {
        if let Some(spider) = self.spider.clone() {
            let spider = spider.clone();
            Engine {
                downloader: self.downloader.clone(),
                scheduler: self.scheduler.clone(),
                crawling: vec![],
                max: self.max,
                spider,
                pipelines: self.pipelines.clone(),
                dmws: self.dmws.clone(),
            }
        } else {
            panic!("必须提供 spider!");
        }
    }
}

pub struct Engine<M, ItemType>
where
    M: Send + Sync + 'static,
    ItemType: Send + Sync + 'static,
{
    downloader: Downloader,
    scheduler: Arc<Mutex<Scheduler<M, ItemType>>>,
    crawling: Vec<tokio::task::JoinHandle<()>>,
    max: usize,
    spider: Arc<dyn Spider<M = M, ItemType = ItemType>>,
    pipelines: Arc<Vec<tokio::sync::Mutex<Box<dyn Pipeline<ItemType>>>>>,
    dmws: Arc<Vec<tokio::sync::Mutex<Box<dyn DownloaderMiddleware>>>>,
}

impl<M, ItemType> Engine<M, ItemType>
where
    M: Send + Sync + 'static,
    ItemType: Send + Sync + 'static,
{
    pub fn builder() -> EngineBuilder<M, ItemType> {
        EngineBuilder::new()
    }

    pub fn new(
        max: usize,
        spider: Arc<dyn Spider<M = M, ItemType = ItemType>>,
        pipelines: Arc<Vec<tokio::sync::Mutex<Box<dyn Pipeline<ItemType>>>>>,
        dmws: Arc<Vec<tokio::sync::Mutex<Box<dyn DownloaderMiddleware>>>>,
    ) -> Self {
        Self {
            downloader: Downloader::new(None),
            scheduler: Arc::new(Mutex::new(Scheduler::<M, ItemType>::new())),
            crawling: vec![],
            max,
            spider,
            pipelines,
            dmws,
        }
    }

    fn init_tasks(&self) {
        let tasks = self.spider.start_tasks();
        for task in tasks {
            self.scheduler.lock().unwrap().enqueue(task);
        }
    }

    pub async fn run(&mut self) {
        // init request
        self.init_tasks();

        loop {
            // 结束条件
            if self.crawling.is_empty() && self.scheduler.lock().unwrap().is_empty() {
                break;
            }

            // 尽量满车出发
            while self.crawling.len() < self.max {
                match self.scheduler.lock().unwrap().dequeue() {
                    Some(task) => {
                        let dl = self.downloader.clone();
                        self.crawling.push(tokio::spawn(Self::crawl(
                            task,
                            dl,
                            self.spider.clone(),
                            self.scheduler.clone(),
                            self.pipelines.clone(),
                            self.dmws.clone(),
                        )));
                    }
                    None => break,
                }
            }

            while let Some(crawler) = self.crawling.pop() {
                crawler.await.unwrap();
            }
        }
    }

    /// 下载
    /// parse
    /// add new reqs
    /// pipelines
    async fn crawl(
        task: Task<M, ItemType>,
        dl: Downloader,
        spider: Arc<dyn Spider<M = M, ItemType = ItemType>>,
        scheduler: Arc<Mutex<Scheduler<M, ItemType>>>,
        pipelines: Arc<Vec<tokio::sync::Mutex<Box<dyn Pipeline<ItemType>>>>>,
        dmws: Arc<Vec<tokio::sync::Mutex<Box<dyn DownloaderMiddleware>>>>,
    ) {
        // let Task { req, meta, parser } = task;

        let new_tasks =
            Self::process_one_task(task, dl, spider, pipelines.clone(), dmws.clone()).await;
        scheduler.lock().unwrap().extends(new_tasks);
    }

    async fn process_one_task(
        task: Task<M, ItemType>,
        dl: Downloader,
        spider: Arc<dyn Spider<M = M, ItemType = ItemType>>,
        pipelines: Arc<Vec<tokio::sync::Mutex<Box<dyn Pipeline<ItemType>>>>>,
        dmws: Arc<Vec<tokio::sync::Mutex<Box<dyn DownloaderMiddleware>>>>,
    ) -> Vec<Task<M, ItemType>> {
        let Task { req, meta, parser } = task;
        match Self::download_one(req.clone(), dl, dmws).await {
            // 如果是 ::Request, 直接重新调度该请求
            IntermediateProcessResult::Request(request) => vec![Task {
                req: request,
                meta,
                parser,
            }],
            // 如果是 ::Response, 交给 spider 解析器
            IntermediateProcessResult::Response(response) => match parser {
                // parse
                Some(parser) => {
                    let ParseResult { tasks, items } = parser.call(response, meta).await;
                    // pipline chain
                    for item in items {
                        Self::pipeline_chain(item, pipelines.clone()).await;
                    }
                    tasks
                }
                None => {
                    let ParseResult { tasks, items } = spider.parse(response, meta).await;
                    for item in items {
                        Self::pipeline_chain(item, pipelines.clone()).await;
                    }
                    tasks
                }
            },
            IntermediateProcessResult::Ignore => vec![],
        }
    }

    async fn download_one(
        req: Request,
        dl: Downloader,
        dmws: Arc<Vec<tokio::sync::Mutex<Box<dyn DownloaderMiddleware>>>>,
    ) -> IntermediateProcessResult {
        let mut req = req;
        // 下载前先经过 DownloaderMiddleware.process_request 处理
        for dm in dmws.iter() {
            let mut dm = dm.lock().await;
            match dm.process_request(req).await {
                // 重新调度
                DownloaderMwResult::FinalRequest(r) => {
                    debug!("重新调度请求 {:?}", r);
                    return IntermediateProcessResult::Request(r);
                }
                // 不下载了, 直接返回 response
                DownloaderMwResult::Response(r) => {
                    debug!("不通过 aruspider 下载, 其他方式返回 response.");
                    return IntermediateProcessResult::Response(r);
                }
                // 继续使用下载器中间件处理
                DownloaderMwResult::IntermediateRequest(r) => req = r,
                // 丢弃
                DownloaderMwResult::Ignore => {
                    debug!("丢弃请求");
                    return IntermediateProcessResult::Ignore;
                }
            }
        }

        let request = req.clone();
        // 下载
        match dl.download(request).await {
            Ok(response) => {
                // 下载后继续使用下载中间件处理
                Self::downloader_middleware_response_chain(&req, response, dmws.clone()).await
            }
            Err(e) => {
                for dm in dmws.iter() {
                    let mut reprocess_response = None; // TODO: Using nested process and holding the response in the outer scope avoids acquiring another lock before releasing the current one.
                    {
                        let mut dm = dm.lock().await;
                        match dm.process_error(&req, &e).await {
                            DownloaderMwErrResult::Continue => (),
                            DownloaderMwErrResult::Request(r) => {
                                return IntermediateProcessResult::Request(r)
                            }
                            DownloaderMwErrResult::Response(r) => reprocess_response = Some(r),
                        }
                    }
                    if let Some(r) = reprocess_response {
                        return Self::downloader_middleware_response_chain(&req, r, dmws.clone())
                            .await;
                    }
                }
                IntermediateProcessResult::Ignore
            }
        }
    }

    async fn downloader_middleware_response_chain(
        request: &Request,
        response: Response,
        dmws: Arc<Vec<tokio::sync::Mutex<Box<dyn DownloaderMiddleware>>>>,
    ) -> IntermediateProcessResult {
        let mut response = response;
        for dm in dmws.iter() {
            let mut dm = dm.lock().await;
            match dm.process_response(request, response).await {
                DownloaderMwResult::FinalRequest(r)
                | DownloaderMwResult::IntermediateRequest(r) => {
                    return IntermediateProcessResult::Request(r);
                }
                DownloaderMwResult::Response(r) => response = r,
                DownloaderMwResult::Ignore => return IntermediateProcessResult::Ignore,
            }
        }
        IntermediateProcessResult::Response(response)
    }

    /// 处理 item 的 pipelines chain
    async fn pipeline_chain(
        item: ItemType,
        pipelines: Arc<Vec<tokio::sync::Mutex<Box<dyn Pipeline<ItemType>>>>>,
    ) {
        let mut item = item;
        for pl in pipelines.iter() {
            let mut pl = pl.lock().await;
            match pl.process(item).await {
                ItemProduct::Item(item_) => item = item_,
                ItemProduct::Ignore => break,
            }
        }
    }
}
